package climspace.hadoop;

public class ArchiveStationImpl implements ArchiveStation {
	private String hostName = "localhost";
	private String hdfsPort = "9000";
	private String mapredPort = "9001";
	
	// hdfs input/output path default here
	private String inputPath = "mesowest/";
	private String outputPath = "mesowest_output/";
	private String otherMesonetPath = "mesonets/";
	
	public ArchiveStationImpl() {
		// use default hostname and ports
	}
	
	/**
	 * ArchiveStationImpl constructor
	 * @param hostName
	 * @param hdfsPort
	 * @param mapredPort
	 */
	public ArchiveStationImpl(String hostName, String hdfsPort, String mapredPort) {
		this.hostName = hostName;
		this.hdfsPort = hdfsPort;
		this.mapredPort = mapredPort;
	}
	
	@Override
	public void uploadDataFile(String fileName) {
		UploadToHDFS upload = new UploadToHDFS(hostName, hdfsPort);
		upload.setDestinationFolder(inputPath);
		upload.copyLocalToRemote(fileName);
	}
	
	@Override
	public int submitHistoricalDataJob(String station, String fromDate, String toDate, String jobID) {
		int status = -1; // error code
		GetHistoricalData getData = new GetHistoricalData(hostName, hdfsPort, mapredPort);
		try {
			status = getData.runAsyncJob(inputPath, outputPath + jobID, station, fromDate, toDate);
		} catch (Exception e) {
			e.printStackTrace();
		}
		return status;
	}

	@Override
	public String downloadDataFileByJobID(String jobID, String outFolder) {
		DownloadFromHDFS download = new DownloadFromHDFS(hostName, hdfsPort);
		String sourceFile = outputPath + jobID + "/part-00000";
		String localFile = outFolder + "/" + jobID + ".out";
		download.copyRemoteToLocal(sourceFile, localFile);
		return localFile;
	}

	public String getHostName() {
		return hostName;
	}

	public void setHostName(String hostName) {
		this.hostName = hostName;
	}

	public String getHdfsPort() {
		return hdfsPort;
	}

	public void setHdfsPort(String hdfsPort) {
		this.hdfsPort = hdfsPort;
	}

	public String getMapredPort() {
		return mapredPort;
	}

	public void setMapredPort(String mapredPort) {
		this.mapredPort = mapredPort;
	}

	public String getInputPath() {
		return inputPath;
	}

	public void setInputPath(String inputPath) {
		this.inputPath = inputPath;
	}

	public String getOutputPath() {
		return outputPath;
	}

	public void setOutputPath(String outputPath) {
		this.outputPath = outputPath;
	}

	@Override
	public void submitHistoricalDataByDateJob(String fromDate, String toDate, String jobID) {
		GetHistoricalDataByDate getData = new GetHistoricalDataByDate(hostName, hdfsPort, mapredPort);
		try {
			getData.run(inputPath, outputPath, fromDate, toDate);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	@Override
	public void submitHistoricalDataByStationJob(String station, String jobID) {
		GetHistoricalDataByStation getData = new GetHistoricalDataByStation(hostName, hdfsPort, mapredPort);
		try {
			getData.run(inputPath, outputPath, jobID);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	@Override
	public void uploadDataFileMesonet(String fileName, String mesonet) {
		UploadToHDFS upload = new UploadToHDFS(hostName, hdfsPort);
		upload.setDestinationFolder(otherMesonetPath + "/" + mesonet);
		upload.copyLocalToRemote(fileName);		
	}
}
